package storage

import (
    "cing.io/cing-api/task"
    "context"
    "fmt"
    log "github.com/sirupsen/logrus"
    clientv3 "go.etcd.io/etcd/client/v3"
    "google.golang.org/protobuf/proto"
)

func PutTask(task *task.Task) error {
    key := getTaskKey(task.Namespace, task.Name)

    bytes, err := proto.Marshal(task)
    if err != nil {
        return err
    }

    _, err = Etcd().Put(context.TODO(), key, string(bytes))
    if err != nil {
        return err
    }
    return nil
}

func GetTask(namespace, name string) *task.Task {
    key := getTaskKey(namespace, name)

    resp, err := Etcd().Get(context.TODO(), key)
    if err != nil {
        return nil
    }

    tsk := &task.Task{}
    err = proto.Unmarshal(resp.Kvs[0].Value, tsk)
    if err != nil {
        return nil
    }
    return tsk
}

func ListTask(namespace string) []*task.Task {
    key := getTaskKey(namespace, "")

    resp, err := Etcd().Get(context.TODO(), key, clientv3.WithPrefix())
    if err != nil {
        return nil
    }

    tasks := make([]*task.Task, resp.Count)
    for i := range tasks {
        tsk := &task.Task{}
        err = proto.Unmarshal(resp.Kvs[i].Value, tsk)
        if err != nil {
            log.Error("Unmarshal error: ", err)
        }
        tasks[i] = tsk
    }
    return tasks

    //tsk1 := &task.Task{
    //    Name:           "a",
    //    Namespace:      namespace,
    //    TaskType:       "FixedRate",
    //    SingerService:  "cing-demo-svc",
    //    SingerSelector: "OneRandomSinger",
    //    Spec: map[string]string{
    //        "schedule":   "3",
    //        "retryLimit": "5",
    //    },
    //}
    //tsk2 := &task.Task{
    //    Name:           "a2",
    //    Namespace:      namespace,
    //    TaskType:       "FixedRate",
    //    SingerService:  "cing-demo-svc",
    //    SingerSelector: "OneRandomSinger",
    //    Spec: map[string]string{
    //        "schedule":   "3",
    //        "retryLimit": "5",
    //    },
    //}
    //return []*task.Task{tsk1, tsk2}
}

func DeleteTask(namespace, name string) error {
    key := getTaskKey(namespace, name)

    _, err := Etcd().Delete(context.TODO(), key)
    if err != nil {
        return err
    }
    return nil
}

func getTaskKey(namespace, name string) string {
    return fmt.Sprintf("%s/%s/%s", taskPrefix, namespace, name)
}
